package com.my.transaction;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
//事务补偿流程
public class Producer2 {

    public static void main(String[] args) throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        producer.setNamesrvAddr("localhost:9876");
        producer.setTransactionListener(new TransactionListener() {//配置事务监听器
            @Override//执行本地事务，对应正常事务流程
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("模拟执行本地事务...");
                return LocalTransactionState.UNKNOW;//状态：未知，超时后会进行事务补偿
            }
            @Override//检查本地事务，对应事务补偿流程
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("检查事务流程...");
                double v = Math.random();
                if(v>=0.5) {
                    System.out.println("事务已提交");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else {
                    System.out.println("事务已回滚");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        });
        producer.start();

        String msg = "hello world transaction";
        Message message = new Message("topic12", "tag1", msg.getBytes());
        //发送事务消息
        TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
        System.out.println(transactionSendResult);

        //千万别关生产者，需要等待本地事务结果
        //producer.shutdown();
    }

}
